package cn.itcast.flink.window.time;
import org.apache.commons.lang3.time.FastDateFormat;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
/**
* 窗口统计案例演示：滚动事件时间窗口（Tumbling EventTime Window），窗口内数据进行词频统计
*/
public class StreamEventTimeWindowApply {
public static void main(String[] args) throws Exception {
// 1. 执行环境-env
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
// TODO: step1. 设置基于事件时间窗口统计
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
// 2. 数据源-source
SingleOutputStreamOperator<String> inputStream = env
.socketTextStream("node1.itcast.cn", 9999)
.filter(line -> null != line && line.trim().split(",").length == 3);
/*
数据格式：1000,a,3 2000,b,2 5000,a,9
*/
// TODO: step2. 指定事件时间EventTime字段
SingleOutputStreamOperator<String> timeStream = inputStream.assignTimestampsAndWatermarks(
new BoundedOutOfOrdernessTimestampExtractor<String>(Time.seconds(0)) { @Override
public long extractTimestamp(String line) {
String[] array = line.trim().split(",");
return Long.parseLong(array[0]);
} }
);
// 3. 数据转换-transformation
SingleOutputStreamOperator<Tuple2<String, Integer>> tupleStream = timeStream.map(
new MapFunction<String, Tuple2<String, Integer>>() { @Override
public Tuple2<String, Integer> map(String line) throws Exception {
String[] array = line.trim().split(",");
return Tuple2.of(array[1], Integer.parseInt(array[2]));
} }
);
// TODO: 事件时间窗口设置
FastDateFormat dateFormat = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss:SSS");
SingleOutputStreamOperator<Tuple2<String, Integer>> sumStream = tupleStream
// 按照单词分组
.keyBy(0)
// TODO: step3. 设置事件时间窗口大小为5秒
.window(TumblingEventTimeWindows.of(Time.seconds(5))) // .timeWindow(Time.seconds(5))
// 窗口内聚合操作: 使用apply函数
.apply(new WindowFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, Tuple, TimeWindow>() { @Override
public void apply(Tuple tuple,
TimeWindow window,
Iterable<Tuple2<String, Integer>> input,
Collector<Tuple2<String, Integer>> out) throws Exception {
// 获取窗口开始start和结束end
String windowStart = dateFormat.format(window.getStart());
String windowEnd = dateFormat.format(window.getEnd());
String word = tuple.toString() ;
// 对窗口内数据进行聚合操作：累加和
int sum = 0 ;
for(Tuple2<String, Integer> item: input){
sum += item.f1 ; }
// 输出内容
String output = "window[" + windowStart + " ~ " + windowEnd + "] -> " + word;
out.collect(Tuple2.of(output, sum));
}
});
// 4. 数据终端-sink
sumStream.printToErr();
/*
测试数据：
1000,a,1
2000,a,1
5000,a,1 --> 触发窗口计算，此条数据不包含
9999,a,1 --> 触发窗口计算，此条数据包含
11000,a,2
14000,b,1
14999,b,1 --> 触发窗口计算，此条数据包含
输出结果：
(window[1970-01-01 08:00:00:000 ~ 1970-01-01 08:00:05:000] -> (a),2)
(window[1970-01-01 08:00:05:000 ~ 1970-01-01 08:00:10:000] -> (a),2)
(window[1970-01-01 08:00:10:000 ~ 1970-01-01 08:00:15:000] -> (a),2)
(window[1970-01-01 08:00:10:000 ~ 1970-01-01 08:00:15:000] -> (b),2)
*/
// 5. 触发执行-execute
env.execute(StreamEventTimeWindowApply.class.getSimpleName());
} }